package com.googlecode.jbeansdb;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.googlecode.jbeansdb.impl.DefaultRequestProcessor;
import com.googlecode.jbeansdb.impl.TcStorageEngine;

/**
 * The main daemon of the beansdb.
 * 
 * @author xumingming64398966
 */
public class Beansdb implements Runnable {
	private static final Logger log = Logger.getLogger(Beansdb.class);
	private StorageEngine engine = new TcStorageEngine();
	/**
	 * the channel on which we will accept connections.
	 */
	private ServerSocketChannel ssc;
	/**
	 * the selector we will be monitoring.
	 */
	private Selector selector;
	/**
	 * the host to listen on.
	 */
	private String host;
	/**
	 * the port to listen on.
	 */
	private int port;

	/**
	 * the worker threads pool.
	 */
	private volatile RequestProcessor processor;
	
	private Map<SocketChannel, List<ByteBuffer>> pendingData = new HashMap<SocketChannel, List<ByteBuffer>>();
	private List<ChangeRequest> pendingChanges = new LinkedList<ChangeRequest>();
	
	private ByteBuffer buffer = ByteBuffer.allocate(1024);
	
	public Beansdb(String host, int port) {
		this.host = host;
		this.port = port;
		
		try {
			init();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public void run() {
		while (true) {
			try {
				// Process any pending changes
				synchronized (this.pendingChanges) {
					Iterator<ChangeRequest> changes = this.pendingChanges.iterator();
					while (changes.hasNext()) {
						log.info("+++++++++++++++++++");
						ChangeRequest change = (ChangeRequest) changes.next();
						switch (change.type) {
						case ChangeRequest.CHANGEOPS:
							SelectionKey key = change.socket
									.keyFor(this.selector);
							key.interestOps(change.ops);
							change.socket.register(this.selector, change.ops);
							log.info("================================================SET " + change.socket + "'s selection key to :" + change.ops);
						}
					}
					this.pendingChanges.clear();
				}

				// Wait for an event one of the registered channels
				this.selector.select();

				// Iterate over the set of keys for which events are available
				Iterator<SelectionKey> selectedKeys = this.selector.selectedKeys().iterator();
				while (selectedKeys.hasNext()) {
					SelectionKey key = (SelectionKey) selectedKeys.next();
					selectedKeys.remove();

					if (!key.isValid()) {
						continue;
					}

					// Check what event is available and deal with it
					if (key.isAcceptable()) {
						this.accept(key);
					} else if (key.isReadable()) {
						this.read(key);
					} else if (key.isWritable()) {
						this.write(key);
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	public void init() throws IOException {
		log.info("starting jbeansdb ...");
		this.selector = SelectorProvider.provider().openSelector();

		// create a nonblocking server socket channel.
		this.ssc = ServerSocketChannel.open();
		this.ssc.configureBlocking(false);
		
		// bind the server socket to the specified address and port.
		this.ssc.socket().bind(new InetSocketAddress(host, port));

		// register the server socket channel, indicating an
		// interest in accepting new connections.
		this.ssc.register(selector, SelectionKey.OP_ACCEPT);
		this.processor = new DefaultRequestProcessor(this);
		log.info("jbeansdb started successfully.");
	}

	private void read(SelectionKey key) {
		SocketChannel channel = (SocketChannel) key.channel();
		String clientKey = channel.socket().getInetAddress().getHostName() + ":" + channel.socket().getPort();
		log.info("New data comes ... " + clientKey);
		
		SocketChannel socketChannel = (SocketChannel) key.channel();
		buffer = ByteBuffer.allocate(1024);
		int numRead;
		try {
			numRead = socketChannel.read(buffer);
		} catch (IOException e) {
			key.cancel();
			try {
				socketChannel.close();
			} catch (IOException e1) {
				e1.printStackTrace();
			}
			return;
		}
		
		if (numRead == -1) {
			// remote entity shut down the socket down cleanly.
			// do the same from our end and cannel the channel.
			try {
				key.channel().close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			key.cancel();
			return;
		}
		
		byte[] data = buffer.array();
		if (data.length == 0) {
			log.error("NO DATA IN THE MESSAGE");
			return;
		} else {
			log.info("Message: " + new String(data) + ", message length: " + numRead);
		}
		
		try {
			processor.process(this, socketChannel, data, numRead);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void send(SocketChannel socket, byte[] data) {
		synchronized (this.pendingChanges) {
			// Indicate we want the interest ops set changed
			this.pendingChanges.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));

			// And queue the data we want written
			synchronized (this.pendingData) {
				List<ByteBuffer> queue = (List<ByteBuffer>) this.pendingData.get(socket);
				if (queue == null) {
					queue = new ArrayList<ByteBuffer>();
					this.pendingData.put(socket, queue);
				}
				queue.add(ByteBuffer.wrap(data));
			}
		}

		// Finally, wake up our selecting thread so it can make the required changes
		this.selector.wakeup();
	}
	
	private void accept(SelectionKey key) throws IOException,
			ClosedChannelException {
		// for an accept to be pending, the channel must be a server socket channel.
		ServerSocketChannel channel = (ServerSocketChannel) key.channel();
		
		// accept the channel and make it nonblocking.
		SocketChannel socketChannel = channel.accept();
		log.info("New connection comes ... " + socketChannel.socket().getRemoteSocketAddress());
		socketChannel.configureBlocking(false);
		
		// register the read event.
		socketChannel.register(selector, SelectionKey.OP_READ);
	}
	
	private void write(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();

		synchronized (this.pendingData) {
			List<ByteBuffer> queue = (List<ByteBuffer>) this.pendingData.get(socketChannel);

			// Write until there's not more data ...
			while (!queue.isEmpty()) {
				ByteBuffer buf = (ByteBuffer) queue.get(0);
				socketChannel.write(buf);
//				if (buf.remaining() > 0) {
//					// ... or the socket's buffer fills up
//					break;
//				}
				queue.remove(0);
			}

			if (queue.isEmpty()) {
				// We wrote away all data, so we're no longer interested
				// in writing on this socket. Switch back to waiting for
				// data.
				key.interestOps(SelectionKey.OP_READ);
			}
		}
	}
	
	public StorageEngine getEngine() {
		return engine;
	}

	public void setEngine(StorageEngine engine) {
		this.engine = engine;
	}

	/**
	 * close the socket, channel and selector.
	 * TODO when should this method be called?
	 * @throws IOException
	 */
	public void shutdown() throws IOException {
		this.ssc.close();
		this.selector.close();
		//this.workers.shutdown();
		this.engine.close();
	}
}
